package com.ybliang.plat.common.rpc.provider.netty;

import com.ybliang.plat.common.rpc.core.RpcRequest;
import com.ybliang.plat.common.rpc.core.RpcResponse;
import com.ybliang.plat.common.rpc.protocol.RpcDecoder;
import com.ybliang.plat.common.rpc.protocol.RpcEncoder;
import com.ybliang.plat.common.rpc.protocol.json.JSONSerializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * @author liangyb1
 */
@Component
public class NettyServer implements ApplicationContextAware, InitializingBean {

    public static Logger LOGGER = LoggerFactory.getLogger(NettyServer.class);
    private static final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    private static final EventLoopGroup workerGroup = new NioEventLoopGroup(4);

    private Map<String, Object> serviceMap = new HashMap<>();

    @Value("${plat.rpc.server.port}")
    private Integer rpcServerPort;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        LOGGER.info("NettyServer setApplicationContext...");
        try {
            Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(Service.class);
            LOGGER.info("NettyServer setApplicationContext... size:{}", serviceBeanMap.size());
            for (Object serviceBean : serviceBeanMap.values()) {
                Class<?> clazz = serviceBean.getClass();
                Class<?>[] interfaces = clazz.getInterfaces();
                for (Class<?> inter : interfaces){
                    String interfaceName = inter.getName();
                    LOGGER.info("加载服务类: {}", interfaceName);
                    serviceMap.put(interfaceName, serviceBean);
                    LOGGER.info("NettyServer assem serviceMap interfaceName:{}, serviceBean:{}", interfaceName, serviceBean);
                }
            }
        } catch (Exception e) {
            LOGGER.error("NettyServer assem serviceMap exception...", e);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        startRpcServerThread();
    }

    private void startRpcServerThread() {
        Thread rpcServerThread = new Thread(() -> {
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap
                        .group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG,1024)
                        .childOption(ChannelOption.SO_KEEPALIVE,true)
                        .childOption(ChannelOption.TCP_NODELAY,true)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                //添加解码器
                                channel.pipeline().addLast(new RpcEncoder(RpcResponse.class, new JSONSerializer()));
                                //添加编码器
                                channel.pipeline().addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                                channel.pipeline().addLast(new NettyServerHandler(serviceMap));
                            }
                        });
                ChannelFuture cf = serverBootstrap.bind(rpcServerPort).sync();
                cf.channel().closeFuture().sync();
//                bind(serverBootstrap, 16789);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }

        });
        rpcServerThread.setName("rpc-server-thread");
        rpcServerThread.setDaemon(true);
        rpcServerThread.start();
        LOGGER.info("NettyServer RpcServerThread started successful in port:{}", rpcServerPort);
    }

    /**
     * 如果端口绑定失败，端口数+1,重新绑定
     *
     * @param serverBootstrap
     * @param port
     */
    public void bind(final ServerBootstrap serverBootstrap,int port) {
        serverBootstrap.bind(port).addListener(future -> {
            if (future.isSuccess()) {
                LOGGER.info("端口[ {} ] 绑定成功",port);
            } else {
                LOGGER.error("端口[ {} ] 绑定失败", port);
//                bind(serverBootstrap, port + 1);
            }
        });
    }


}
